package com.spzx.common.rabbitmq.configure;

import com.alibaba.fastjson2.JSON;
import com.spzx.common.rabbitmq.cons.SpzxMqCons;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.concurrent.TimeUnit;

@Configuration
@Slf4j
public class SpzxRabbitmqConfig {
    //配置生产者确认回调配置
    @Resource
    RabbitTemplate rabbitTemplate;
    @Resource
    RedisTemplate redisTemplate;
    @PostConstruct
    public void init(){
        //2、消息未到达队列的回调
        rabbitTemplate.setReturnsCallback((ReturnedMessage returned)->{
            System.out.println("消息未到达队列："+ JSON.toJSONString(returned));
            // returned.message.messageProperties.headers.spring_returned_message_correlation
            //直接获取消息内容重新发送
            retrySendMsg(returned.getMessage().getMessageProperties()
                         .getHeader("spring_returned_message_correlation")
                        );
        });
        //1、消息是否到达交换机的回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause)->{
            if(ack){
                //消息到达交换机
                //                System.out.println("消息到达交换机成功");
            }else{
                //                System.out.println("消息到达交换机失败："+cause);
                retrySendMsg(correlationData.getId());
            }
        });
    }

    private void retrySendMsg(String correlationDataId) {
        //1、读取重试消息内容
        Object obj = redisTemplate.opsForValue().get(correlationDataId);
        if(obj==null){
            //没有获取到重试消息内容
            return;
        }
        SpzxCorrelationData spzxCorrelationData = (SpzxCorrelationData) obj;
        //2、判断是否需要重试
        if(!spzxCorrelationData.isRetry()){
            redisTemplate.delete(spzxCorrelationData.getId());//删除重试缓存
            //不需要重试
            return;
        }
        //3、判断重试次数是否已经达到阈值
        if(spzxCorrelationData.getCount()>=spzxCorrelationData.getRetryLimit()){
            //重试次数达到阈值：不再重试
            //保存消息内容到日志文件中/到日志表中
            log.error("消息重试{}次,超过阈值{},发送失败，消息内容：{}" ,
                      spzxCorrelationData.getCount() ,spzxCorrelationData.getRetryLimit() ,
                      JSON.toJSONString(spzxCorrelationData.getMessage()));
            redisTemplate.delete(spzxCorrelationData.getId());//删除重试缓存
            return;
        }
        //4、重试发送消息
        if(spzxCorrelationData.getTimeIntervel()>0){
            try {
                TimeUnit.MILLISECONDS.sleep(spzxCorrelationData.getTimeIntervel());
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        //5、更新消息发送的次数
        spzxCorrelationData.setCount(spzxCorrelationData.getCount()+1);
        rabbitTemplate.convertAndSend(spzxCorrelationData.getExchange(),spzxCorrelationData.getRoutingKey(),
                                      spzxCorrelationData.getMessage()
                                      ,spzxCorrelationData);

        //同步到redis中
        redisTemplate.opsForValue().set(spzxCorrelationData.getId(),spzxCorrelationData);
    }


    //死信队列死信交换机绑定
    @Bean
    public Exchange spzxOrderDeadExchange(){
        return ExchangeBuilder.topicExchange(SpzxMqCons.ORDER_SUBMIT_DEAD_EXCHANGE)
            .durable(true)
            .build();
    }
    @Bean
    public Queue spzxOrderSubmitDeadQueue(){
        return QueueBuilder
            .durable(SpzxMqCons.ORDER_SUBMIT_DEAD_QUEUE)
            .build();
    }
    @Bean
    public Binding spzxOrderSubmitDeadBinding(Exchange spzxOrderDeadExchange,
                                              Queue spzxOrderSubmitDeadQueue){
        return BindingBuilder.bind(spzxOrderSubmitDeadQueue)
            .to(spzxOrderDeadExchange)
            .with(SpzxMqCons.ORDER_SUBMIT_DEAD_MSG_ROUTING_KEY)
            .noargs();
    }
    //创建测试业务队列业务交换机绑定
    //给测试业务队列绑定死信队列
    //    @Bean
    //    public Exchange spzxTestExchange(){
    //        return ExchangeBuilder.topicExchange("spzx.mq.test1.exchange")
    //                .durable(true)
    //                .build();
    //    }
    //
    //    @Bean
    //    public Queue spzxTestQueue(){
    //        //当该队列中的消息变为死信以后，消息会通过死信交换机分发，路由key为死信路由key
    //        return QueueBuilder
    //                .durable("spzx.mq.test1.queue")
    //                .deadLetterExchange("spzx.mq.test1.dead.exchange") //给业务队列设置死信交换机
    //                .deadLetterRoutingKey("spzx.mq.test1.dead") //给业务队列设置死信路由key
    //                .build();
    //    }
    //    @Bean
    //    public Binding spzxTestBinding(Exchange spzxTestExchange,
    //                                       Queue spzxTestQueue){
    //        return BindingBuilder.bind(spzxTestQueue)
    //                .to(spzxTestExchange)
    //                .with("a.#")
    //                .noargs();
    //    }



    @Bean
    public Exchange spzxOrderExchange(){
        return ExchangeBuilder.topicExchange(SpzxMqCons.ORDER_EXCHANGE)
            .durable(true)
            .build();
    }
    // 延迟关单的业务队列
    @Bean
    public Queue spzxOrderQueue(){
        //当该队列中的消息变为死信以后，消息会通过死信交换机分发，路由key为死信路由key
        return QueueBuilder
            .durable(SpzxMqCons.ORDER_SUBMIT_QUEUE)
            .ttl(SpzxMqCons.ORDER_SUBMIT_DELAY_TIME)
            .deadLetterExchange(SpzxMqCons.ORDER_SUBMIT_DELAY_EXCHANGE) //给业务队列设置延迟交换机
            .deadLetterRoutingKey(SpzxMqCons.ORDER_SUBMIT_DELAY_MSG_ROUTING_KEY) //给业务队列设置延迟路由key
            .build();
    }
    @Bean
    public Binding spzxOrderBinding(Exchange spzxOrderExchange,
                                    Queue spzxOrderQueue){
        return BindingBuilder.bind(spzxOrderQueue)
            .to(spzxOrderExchange)
            .with(SpzxMqCons.ORDER_SUBMIT_ROUTING_KEY)
            .noargs();
    }
    //释放库存的 队列和绑定
    @Bean
    public Queue spzxStockReleaseQueue(){
        //当该队列中的消息变为死信以后，消息会通过死信交换机分发，路由key为死信路由key
        return QueueBuilder
            .durable(SpzxMqCons.SPZX_STOCK_RELEASE_QUEUE)
            .build();
    }
    @Bean
    public Binding spzxStockReleaseBinding(Exchange spzxOrderExchange,
                                           Queue spzxStockReleaseQueue){
        return BindingBuilder.bind(spzxStockReleaseQueue)
            .to(spzxOrderExchange)
            .with(SpzxMqCons.SPZX_STOCK_RELEASE_ROUTING_KEY)
            .noargs();
    }



    //延迟队列  延迟交换机 绑定： 用来接收 SpzxMqCons.ORDER_SUBMIT_QUEUE过期的消息
    @Bean
    public Exchange spzxOrderDelayExchange(){
        return ExchangeBuilder.topicExchange(SpzxMqCons.ORDER_SUBMIT_DELAY_EXCHANGE)
            .durable(true)
            .build();
    }
    //用来接收30分钟后变为死信的延迟消息，延迟关单消费者获取该队列中的消息进行消费
    @Bean
    public Queue spzxOrderSubmitDelayQueue(){
        return QueueBuilder
            .durable(SpzxMqCons.ORDER_SUBMIT_DELAY_QUEUE)
            .build();
    }
    @Bean
    public Binding spzxOrderSubmitDelayBinding(Exchange spzxOrderDelayExchange,
                                               Queue spzxOrderSubmitDelayQueue){
        return BindingBuilder.bind(spzxOrderSubmitDelayQueue)
            .to(spzxOrderDelayExchange)
            .with(SpzxMqCons.ORDER_SUBMIT_DELAY_MSG_ROUTING_KEY)
            .noargs();
    }
}